package org.apache.hadoop.hive.cassandra.cql;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;

import org.apache.cassandra.thrift.ColumnDef;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.cassandra.CassandraException;
import org.apache.hadoop.hive.cassandra.input.cql.HiveCqlInputFormat;
import org.apache.hadoop.hive.cassandra.output.cql.HiveCqlOutputFormat;
import org.apache.hadoop.hive.cassandra.serde.AbstractCassandraSerDe;
import org.apache.hadoop.hive.cassandra.serde.cql.CqlMapSerde;
import org.apache.hadoop.hive.cassandra.serde.cql.CqlSerDe;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.index.IndexPredicateAnalyzer;
import org.apache.hadoop.hive.ql.index.IndexSearchCondition;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.HiveStoragePredicateHandler;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CqlStorageHandler implements HiveStorageHandler, HiveMetaHook, HiveStoragePredicateHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(CqlStorageHandler.class);

    private Configuration configuration;

    @Override
    public void configureTableJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) {
        Properties tableProperties = tableDesc.getProperties();
        /**
        Iterator<Entry<Object, Object>> it = tableProperties.entrySet().iterator();  
        while (it.hasNext()) {  
            Entry<Object, Object> entry = it.next();  
            Object key = entry.getKey();  
            Object value = entry.getValue();  
            LOGGER.info("tableDesc---key   :" + key+" ---value :" + value);  
        } 
        Set<String> jobPropertiesKeys =  jobProperties.keySet();
        for(String str:jobPropertiesKeys){
        	LOGGER.info("jobProperties---key   :" + str+" ---value :" + jobProperties.get(str)); 
        }
        **/
        
        String test_zyy = tableProperties.getProperty("yangdi");
        System.out.println("test_zyy==="+test_zyy);
        
        //Keyspace
        String keyspace = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_KEYSPACE_NAME);
        if (keyspace == null) {
            //keyspace = tableProperties.getProperty(hive_metastoreConstants.META_TABLE_DB);
        	throw new RuntimeException("cassandra keyspace不能为空");
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_KEYSPACE_NAME, keyspace);

        //Identify ColumnFamily
        String columnFamily = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_CF_NAME);
        if (columnFamily == null) {
            //columnFamily = tableProperties.getProperty(hive_metastoreConstants.META_TABLE_NAME);
        	throw new RuntimeException("cassandra cf不能为空");
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CF_NAME, columnFamily);
        

        //列映射关系
        String columnInfo = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING);
        if(columnInfo == null){
        	columnInfo = CqlSerDe.createColumnMappingString(tableProperties.getProperty(org.apache.hadoop.hive.serde.Constants.LIST_COLUMNS));
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING, columnInfo);
        LOGGER.info("列名:{}",jobProperties.get(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING));
        
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_COL_TYPE_MAPPING,CqlSerDe.createColumnMappingString(tableProperties.getProperty(org.apache.hadoop.hive.serde.Constants.LIST_COLUMN_TYPES)));
        LOGGER.info("类型名称:{}",jobProperties.get(AbstractCassandraSerDe.CASSANDRA_COL_TYPE_MAPPING));
        //cassandra 节点ip列表
        String host = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HOST);
        if (host == null || "".equals(host)) {
            //host = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HOST, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HOST);
        	throw new RuntimeException("cassandra hosts不能为空");
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HOST, host);
        
        //cassandra端口
        String port = configuration.get(AbstractCassandraSerDe.CASSANDRA_PORT);
        if (port == null) {
            port = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_PORT, AbstractCassandraSerDe.DEFAULT_CASSANDRA_PORT);
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PORT, port);

        String maxThreads = configuration.get(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY);
        if (maxThreads == null) {
            maxThreads = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HADOOP_MAX_THREADS);
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HADOOP_MAX_THREADS_KEY, maxThreads);

        String maxRetries = configuration.get(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY);
        if (maxRetries == null) {
            maxRetries = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HADOOP_RETRIES);
        }
        jobProperties.put(AbstractCassandraSerDe.CASSANDRA_HADOOP_RETRIES_KEY, maxRetries);


        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_PARTITIONER) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PARTITIONER,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_PARTITIONER,
                            "org.apache.cassandra.dht.Murmur3Partitioner"));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_PARTITIONER, configuration.get(AbstractCassandraSerDe.CASSANDRA_PARTITIONER));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL,
                            AbstractCassandraSerDe.DEFAULT_CONSISTENCY_LEVEL));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL, configuration.get(AbstractCassandraSerDe.CASSANDRA_CONSISTENCY_LEVEL));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_RANGE_BATCH_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_RANGE_BATCH_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_SLICE_PREDICATE_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_SPLIT_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_SPLIT_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE,
                            Integer.toString(AbstractCassandraSerDe.DEFAULT_BATCH_MUTATION_SIZE)));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE, configuration.get(AbstractCassandraSerDe.CASSANDRA_BATCH_MUTATION_SIZE));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_START));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH, configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_FINISH));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR, ""));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR,
                    configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_COMPARATOR));
        }

        if (configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED) == null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED,
                    tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED, "false"));
        } else {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED,
                    configuration.get(AbstractCassandraSerDe.CASSANDRA_SLICE_PREDICATE_RANGE_REVERSED));
        }

        //Set the indexed column names - leave unset if we have problems determining them
        /**
        String indexedColumns = tableProperties.getProperty(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS);
        if (indexedColumns != null) {
            jobProperties.put(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS, indexedColumns);
        } else {
            try {
                Set<ColumnDef> columns = CqlPushdownPredicate.getIndexedColumns(host, Integer.parseInt(port), keyspace, columnFamily);
                jobProperties.put(AbstractCassandraSerDe.CASSANDRA_INDEXED_COLUMNS, CqlPushdownPredicate.serializeIndexedColumns(columns));
            } catch (CassandraException e) {
                // this results in the property remaining unset on the Jobconf, so indexes will not be used on the C* side
                LOGGER.info("Error determining cassandra indexed columns, will not include in JobConf", e);
            }
        }
        **/

    }

    @Override
    public Class<? extends InputFormat> getInputFormatClass() {
        return HiveCqlInputFormat.class;
    }

    @Override
    public HiveMetaHook getMetaHook() {
        return this;
    }

    @Override
    public Class<? extends OutputFormat> getOutputFormatClass() {
        return HiveCqlOutputFormat.class;
    }

    @Override
    public Class<? extends SerDe> getSerDeClass() {
        //return CqlSerDe.class;
    	return CqlMapSerde.class;
    }

    @Override
    public Configuration getConf() {
        return this.configuration;
    }

    @Override
    public void setConf(Configuration conf) {
    	/**
    	Iterator<Map.Entry<String, String>> iterator = conf.iterator();
    	while(iterator.hasNext()){
    		Map.Entry<String, String> entry = iterator.next();
    		System.out.println("setConf -key=" + entry.getKey() + ":value=" + entry.getValue());
    		LOGGER.info("setConf LOG -key=" + entry.getKey() + ":value=" + entry.getValue());
    	}**/
    	
        this.configuration = conf;
    }

    @Override
    public void preCreateTable(Table table) throws MetaException {
    	/**
        boolean isExternal = MetaStoreUtils.isExternalTable(table);
        if (!isExternal) {
            throw new MetaException("Cassandra tables must be external.");
        }

        if (table.getSd().getLocation() != null) {
            throw new MetaException("LOCATION may not be specified for Cassandra.");
        }

        CqlManager manager = new CqlManager(table);

        try {
            //open connection to cassandra
            manager.openConnection();
            if (!manager.doesKeyspaceExist()) {
                LOGGER.info("Keyspace doesnot exist. Creating keyspace {}", table.getDbName());
                manager.createKeyspace();
            }
            //create the column family if it doesn't exist.
            manager.createCFIfNotFound();
        } finally {
            manager.closeConnection();
        }
        **/
    }

    @Override
    public void commitCreateTable(Table table) throws MetaException {
        // No work needed
    }

    @Override
    public void commitDropTable(Table table, boolean deleteData) throws MetaException {
    	/**
        boolean isExternal = MetaStoreUtils.isExternalTable(table);
        if (deleteData && !isExternal) {
            CqlManager manager = new CqlManager(table);

            try {
                //open connection to cassandra
                manager.openConnection();
                //drop the table
                manager.dropTable();
            } finally {
                manager.closeConnection();
            }
        }
        **/
    }

    @Override
    public void preDropTable(Table table) throws MetaException {
        // nothing to do
    }

    @Override
    public void rollbackCreateTable(Table table) throws MetaException {
        // No work needed
    }

    @Override
    public void rollbackDropTable(Table table) throws MetaException {
        // nothing to do
    }

    @Override
    public HiveAuthorizationProvider getAuthorizationProvider() throws HiveException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public void configureInputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) {
        configureTableJobProperties(tableDesc, jobProperties);

    }

    @Override
    public void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) {

        configureTableJobProperties(tableDesc, jobProperties);

    }

    /**
     * Cassandra requires that an IndexClause must contain at least one
     * IndexExpression with an EQ operator on a configured index column. Other
     * IndexExpression structs may be added to the IndexClause for non-indexed
     * columns to further refine the results of the EQ expression.
     * <p/>
     * In order to push down the predicate filtering, we first get a list of
     * indexed columns. If there are no indexed columns, we can't push down the
     * predicate. We then walk down the predicate, and see if there is any
     * filtering that matches the indexed columns. If there is no matching, we
     * can't push down the predicate. For any matching column that is found, we
     * need to verify that there is at least one equal operator. If there is no
     * equal operator, we can't push down the predicate.
     */
    @Override
    public DecomposedPredicate decomposePredicate(JobConf jobConf, Deserializer deserializer, ExprNodeDesc predicate) {
        try {
            CqlSerDe cassandraSerde = (CqlSerDe) deserializer;
            String host = jobConf.get(AbstractCassandraSerDe.CASSANDRA_HOST, AbstractCassandraSerDe.DEFAULT_CASSANDRA_HOST);
            int port = jobConf.getInt(AbstractCassandraSerDe.CASSANDRA_PORT, Integer.parseInt(AbstractCassandraSerDe.DEFAULT_CASSANDRA_PORT));
            String ksName = cassandraSerde.getCassandraKeyspace();
            String cfName = cassandraSerde.getCassandraColumnFamily();
            Set<ColumnDef> indexedColumns = CqlPushdownPredicate.getIndexedColumns(host, port, ksName, cfName);
            if (indexedColumns.isEmpty()) {
                return null;
            }

            IndexPredicateAnalyzer analyzer = CqlPushdownPredicate.newIndexPredicateAnalyzer(indexedColumns);
            List<IndexSearchCondition> searchConditions = new ArrayList<IndexSearchCondition>();
            ExprNodeDesc residualPredicate = analyzer.analyzePredicate(predicate, searchConditions);

            if (searchConditions.isEmpty()) {
                return null;
            }

            if (!CqlPushdownPredicate.verifySearchConditions(searchConditions)) {
                return null;
            }

            DecomposedPredicate decomposedPredicate = new DecomposedPredicate();
            decomposedPredicate.pushedPredicate = analyzer.translateSearchConditions(searchConditions);
            decomposedPredicate.residualPredicate = (ExprNodeGenericFuncDesc) residualPredicate;

            return decomposedPredicate;
        } catch (CassandraException e) {
            //We couldn't get the indexed column names from Cassandra, return null and let Hive handle the filtering
            LOGGER.info("Error during predicate decomposition", e);
            return null;
        }
    }

	@Override
	public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
		/**
		Iterator<Entry<Object, Object>> it = tableDesc.getProperties().entrySet().iterator();  
        while (it.hasNext()) {  
            Entry<Object, Object> entry = it.next();  
            Object key = entry.getKey();  
            Object value = entry.getValue();  
            System.out.println("configureJobConf-tableDesc---key   :" + key+" ---value :" + value);  
        } 
        Iterator<Map.Entry<String, String>> sdsd = jobConf.iterator();
        while(sdsd.hasNext()){
        	Entry<String, String> entry = sdsd.next();  
            String key = entry.getKey();  
            String value = entry.getValue();  
            System.out.println("configureJobConf-jobConf---key   :" + key+" ---value :" + value); 
        }
        **/
	}

}
